package net.zkbc.helloworld.actor;

import java.io.Serializable;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import akka.persistence.SnapshotOffer;
import akka.persistence.UntypedPersistentActor;

@Component
@Scope("prototype")
public class HitsActor extends UntypedPersistentActor {

	public static class Count implements Serializable {
		private static final long serialVersionUID = 1L;

		private String from;

		public Count(String from) {
			if (StringUtils.isEmpty(from)) {
				this.from = "default";
			} else {
				this.from = from;
			}
		}

		public String getFrom() {
			return from;
		}

		public void setFrom(String from) {
			this.from = from;
		}

	}

	private static final Logger LOG = LoggerFactory.getLogger(HitsActor.class);

	private int count;

	@Override
	public String persistenceId() {
		return "HitsActor-" + getContext().parent().path().name();
	}

	@Override
	public void onReceiveCommand(Object msg) throws Exception {
		if (msg instanceof Count) {
			LOG.info("count={}", count);
			persist(msg, c -> {
				updateState(msg);
			});
		} else if (msg.equals("snap")) {
			saveSnapshot(count);
		} else {
			unhandled(msg);
		}
	}

	@Override
	public void onReceiveRecover(Object msg) throws Exception {
		if (msg instanceof Count) {
			updateState(msg);
		} else if (msg instanceof SnapshotOffer) {
			count = (int) ((SnapshotOffer) msg).snapshot();
		} else {
			unhandled(msg);
		}
	}

	private void updateState(Object msg) {
		count++;
	}
}
